package day08;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import po.OrderEvent;

import java.util.List;
import java.util.Map;

/**
 * @Description: 订单超时
 * @Author ZYX
 * @Date 2022/2/20 19:30
 * @Version 1.0
 */
public class Demo06 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<OrderEvent> orderStream = env.fromElements(
                new OrderEvent("order-1", "create", 1000L),
                new OrderEvent("order-2", "create", 2000L),
                new OrderEvent("order-1", "pay", 3000L),
                new OrderEvent("order-2", "pay", 7100L)
                ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<OrderEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
                            @Override
                            public long extractTimestamp(OrderEvent orderEvent, long l) {
                                return orderEvent.timestamp;
                            }
                        })
        );
        Pattern<OrderEvent, OrderEvent> orderEventPattern = Pattern
                .<OrderEvent>begin("create")
                .where(new SimpleCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent orderEvent) throws Exception {
                        return "create".equalsIgnoreCase(orderEvent.eventType);
                    }
                })
                .next("pay")
                .where(new SimpleCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent orderEvent) throws Exception {
                        return "pay".equalsIgnoreCase(orderEvent.eventType);
                    }
                })
                // 要求两个之间在5s内发生
                .within(Time.seconds(5));

        PatternStream<OrderEvent> patternStream = CEP.pattern(orderStream.keyBy(obj -> obj.orderId), orderEventPattern);

        SingleOutputStreamOperator<String> result = patternStream.flatSelect(
                // 超时事件将发送至侧输出流
                new OutputTag<String>("timeout") {
                },
                new PatternFlatTimeoutFunction<OrderEvent, String>() {
                    @Override
                    public void timeout(Map<String, List<OrderEvent>> map, long l, Collector<String> out) throws Exception {
                        OrderEvent create = map.get("create").get(0);
                        // 发送到侧输出流中
                        out.collect("订单id：" + create.orderId + "的订单超时了");
                    }
                },
                new PatternFlatSelectFunction<OrderEvent, String>() {
                    @Override
                    public void flatSelect(Map<String, List<OrderEvent>> map, Collector<String> out) throws Exception {
                        OrderEvent pay = map.get("pay").get(0);
                        out.collect("订单id：" + pay.orderId + "的订单已支付。");
                    }
                }
        );
        result.print();
        result.getSideOutput(new OutputTag<String>("timeout"){}).print();


        env.execute();
    }

}
